package com.ahui.test;

import com.athui.bean.ColumnMetaData;
import com.athui.bean.DataType;
import com.athui.enums.DataRegularEnums;
import com.athui.service.QueueExplorerService;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.regex.Pattern;

/**
 * @description: TODO
 * @autor: zhangzhonghui
 * @create: 2024-08-20 17:02
 * @Version: 1.0
 */
public class StringQueueExplorerService extends QueueExplorerService<String,String> {

    private List<String> dataList;

    public StringQueueExplorerService(List<String> dataList) {
        super(dataList);
        this.dataList = dataList;
    }

    @Override
    public String transformBySourceType(String input) {
        return input;
    }

    @Override
    public DataStream<String> source(StreamExecutionEnvironment env) {
        return env.fromCollection(dataList).map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return  transformBySourceType(value);
            }
        });
    }

    @Override
    public Collection<ColumnMetaData> getColumnMetaDataList(Collection<ColumnMetaData> collect) {
        ColumnMetaData columnMetaData = new ColumnMetaData();
        columnMetaData.setColumnCode("asadfsdafsas");
        columnMetaData.setColumnName("str");
        columnMetaData.setTypeLabel(DataType.TEXT);
        columnMetaData.setIndex("0");
        columnMetaData.setTextPattern(DataRegularEnums.TEXT_PHONE_REGULAR.getValue());
        columnMetaData.setFlinkColumnType(BasicTypeInfo.STRING_TYPE_INFO);
        collect.add(columnMetaData);

        return collect;
    }


    @Override
    public Object transformByValue(String index,String row) {
        System.out.println(index+":"+row);
        return row;
    }

    @Override
    public String getJobName() {
        return "string-queue-job";
    }

    @Override
    public void sink(DataStream<ColumnMetaData> stream) {
        stream.print();
    }

    public static void main(String[] args) {

        List<String> dataList = new ArrayList<>();
        dataList.add("13000008918");
        dataList.add("13000008917");
        dataList.add("13000008916");
        dataList.add("13000008915");
        dataList.add("13000008914");
        dataList.add("13000008913");
        dataList.add("13000008912");
        dataList.add("13000008911");
        dataList.add("13000008910");
        dataList.add("asgjkaljdcklas");
        dataList.add("31241233414312");

        StringQueueExplorerService explorerService =new  StringQueueExplorerService(dataList);

        try {
            explorerService.run();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


}
